package com.hjzd.ai.smartpractice.websocket;

import com.alibaba.nls.client.AccessToken;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.*;
import com.google.gson.reflect.TypeToken;
import com.hjzd.ai.smartpractice.config.WebClientConfig;
import com.hjzd.ai.smartpractice.constant.LLMRole;
import com.hjzd.ai.smartpractice.constant.ServiceConstant;
import com.hjzd.ai.smartpractice.entity.DTO.chat.ChatCompletionMessage;
import com.hjzd.ai.smartpractice.entity.DTO.chat.ChatResponseThread;
import com.hjzd.ai.smartpractice.service.RedisService;
import com.hjzd.ai.smartpractice.util.ConversationUtil;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.netty.http.client.HttpClient;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import static com.hjzd.ai.smartpractice.constant.ServiceConstant.*;
import static com.hjzd.ai.smartpractice.websocket.SessionManager.*;

public class VolcanoAsrClient extends WebSocketClient {

    private static final Logger logger = LoggerFactory.getLogger(VolcanoAsrClient.class);
    private static final Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss")
            .setLongSerializationPolicy(LongSerializationPolicy.STRING).disableHtmlEscaping().create();
    private WebClient webClient;
    // WebSocket 协议相关常量
    private static final byte PROTOCOL_VERSION = 0b0001;
    private static final byte DEFAULT_HEADER_SIZE = 0b0001;
    private static final byte FULL_CLIENT_REQUEST = 0b0001;
    private static final byte AUDIO_ONLY_REQUEST = 0b0010;
    private static final byte POS_SEQUENCE = 0b0001;
    private static final byte NEG_WITH_SEQUENCE = 0b0011;
    private static final byte JSON = 0b0001;
    private static final byte GZIP = 0b0001;

    private WebSocketSession session;
    private RedisService redisService;
    private WebClientConfig webClientConfig;
    private String ttsUrl;
    private String ttsType;
    private String llmFlag;
    private String saveVoicePath;
    private int sequence = 1; // 全局序列号
    private ChatResponseThread ttsThread;
    private Set<String> sentWords; // 用于跟踪已发送的单词
    private String conversationId; // 用于存储会话 ID
    private String llmUrl; // 用于存储会话 ID
    private StepfunTTSClient ttsClient;
    private AliTtsClient aliTtsClient;
    private boolean hasSentInputTextStart = false;


    public VolcanoAsrClient(URI serverURI, WebSocketSession session, RedisService redisService,
                            WebClientConfig webClientConfig, String ttsUrl, String saveVoicePath, String ttsType
            , String llmFlag, String llmUrl) {
        super(serverURI);
        this.session = session;
        this.redisService = redisService;
        this.webClientConfig = webClientConfig;
        this.ttsUrl = ttsUrl;
        this.ttsType = ttsType;
        this.saveVoicePath = saveVoicePath;
        HttpClient httpClient = HttpClient.create();
        ClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);
        this.webClient = webClientConfig.webClientBuilder()
                .clientConnector(connector)
                .defaultHeader(HttpHeaders.CONTENT_TYPE, "application/json")
                .build();
        // 设置自定义请求头
        this.addHeader("X-Api-App-Key", "1393667404");
        this.addHeader("X-Api-Access-Key", "9eGvcslDRowNgVESTSxg-DUJkEezgVGs");
        this.addHeader("X-Api-Resource-Id", "volc.bigasr.sauc.duration");
        this.addHeader("X-Api-Connect-Id", UUID.randomUUID().toString());
        this.sentWords = new HashSet<>();
        this.llmFlag = llmFlag;
        this.llmUrl = llmUrl;
        if (this.llmFlag.equals("LLM_CUSTOMER")) {
            //提前获取会话ID
            this.conversationId = ConversationUtil.startConversation(this.llmUrl);
        }
        if (ttsType.equals("STEP")) {
            try {
                URI uri = new URI("wss://api.stepfun.com/v1/realtime/audio?model=step-tts-mini");
                this.ttsClient = new StepfunTTSClient(uri, session);
                this.ttsClient.connect();
            } catch (Exception e) {
                logger.error("初始化 Stepfun TTS 客户端失败", e);
            }
        } else if (ttsType.equals("ALI")) {
            try {
                AccessToken accessToken = new AccessToken("LTAI5tBtiXs4SwSjHPHkTDif", "QAkMv0E0uQhXD26KhjRzUna0FPA0k6");
                accessToken.apply();
                String token = accessToken.getToken();
                long expireTime = accessToken.getExpireTime();
                aliTtsClient = new AliTtsClient("NNJtv1mxBChM3RIs", token, session);
            } catch (Exception e) {
                logger.error("开启阿里TTS失败:{}", e);
            }
        }
    }

    @Override
    public void onOpen(ServerHandshake handshakedata) {
        logger.info("Connected to Volcano ASR server");
        sendFullClientRequest();

    }

    @Override
    public void onMessage(ByteBuffer bytes) {
        try {
            byte[] data = new byte[bytes.remaining()];
            bytes.get(data);
            parseServerResponse(data);
        } catch (Exception e) {
            logger.error("处理二进制消息错误", e);
        }
    }

    @Override
    public void onMessage(String message) {
        logger.info("接收到VolcanoAsrClient消息：{}", message);
        try {

            // 处理 ASR 返回的消息
            JsonObject jsonObject = GSON.fromJson(message, JsonObject.class);
            if (jsonObject.has("error")) {
                // 处理错误信息
                String error = jsonObject.get("error").getAsString();
                logger.error("火山 ASR 返回错误: {}", error);
                return;
            }

            String text = jsonObject.get("text").getAsString();
            boolean isFinal = jsonObject.get("is_final").getAsBoolean();

            if (isFinal) {
                // 处理最终结果
                handleFinalResult(session, text);
            }
        } catch (Exception e) {
            logger.error("处理火山 ASR 消息错误", e);
        }
    }


    private void parseServerResponse(byte[] data) {
        try {
            // 解析协议头 -----------------------------------------
            byte protocolVersion = (byte) ((data[0] >> 4) & 0x0F);
            byte headerSize = (byte) (data[0] & 0x0F);
            byte messageType = (byte) ((data[1] >> 4) & 0x0F);
            byte flags = (byte) (data[1] & 0x0F);
            byte serialMethod = (byte) ((data[2] >> 4) & 0x0F);
            byte compressionType = (byte) (data[2] & 0x0F);
            byte reserved = data[3];

            // 解析序列号 -----------------------------------------
            int sequence = ByteBuffer.wrap(data, 4, 4).getInt();

            // 解析 Payload 大小 -----------------------------------
            int payloadSize = ByteBuffer.wrap(data, 8, 4).getInt();

            // 提取 Payload ---------------------------------------
            byte[] payload = new byte[payloadSize];
            System.arraycopy(data, 12, payload, 0, payloadSize);

            // GZIP 解压缩 ----------------------------------------
            if (compressionType == GZIP) {
                payload = gzipDecompress(payload);
                if (payload.length == 0) {
                    logger.warn("解压缩后的 payload 为空");
                    return;
                }
            }

            // 仅处理 JSON 格式的响应 -------------------------------
            if (serialMethod != JSON) {
                logger.warn("非 JSON 格式的响应, serialMethod={}", serialMethod);
                return;
            }

            // 解析 JSON ------------------------------------------
            String jsonStr = new String(payload, StandardCharsets.UTF_8);
            JsonObject jsonObject = GSON.fromJson(jsonStr, JsonObject.class);
            //logger.info("原始服务端响应: {}", jsonStr);

            // 错误处理 -------------------------------------------
            if (jsonObject.has("error")) {
                String error = jsonObject.get("error").getAsString();
                logger.error("ASR 服务端返回错误: {}", error);
                if (error.contains("timeout")) {
                    logger.error("ASR 会话超时: {}", error);
                    reconnectAndResend(); // 触发重连逻辑
                    return;
                }
                return;
            }

            // 提取识别结果 ----------------------------------------
            if (jsonObject.has("result")) {
                JsonObject result = jsonObject.getAsJsonObject("result");
                //logger.info("ASR原始返回：{}",result);
                // 获取文本内容
                String text = "";
                if (result.has("text") && !result.get("text").isJsonNull()) {
                    text = result.get("text").getAsString().trim();
                }
                if (!hasSentInputTextStart) {
                    try {
                        session.sendMessage(new TextMessage(inputTextStart));
                        hasSentInputTextStart = true;
                        logger.info("发送ASR消息 inputTextStart");
                    } catch (IOException e) {
                        logger.error("发送 inputTextStart 消息失败", e);
                    }
                }
                // 判断是否为最终结果
                boolean isFinal = false;
                if (result.has("utterances") && result.get("utterances").isJsonArray()) {
                    JsonArray utterances = result.getAsJsonArray("utterances");
                    for (JsonElement element : utterances) {
                        JsonObject utterance = element.getAsJsonObject();
                        if (utterance.has("words") && utterance.get("words").isJsonArray()) {
                            JsonArray words = utterance.getAsJsonArray("words");
                            // 遍历每个 word
                            for (JsonElement wordElement : words) {
                                JsonObject wordObject = wordElement.getAsJsonObject();
                                String word = wordObject.get("text").getAsString();

                                // 只发送新的单词
                                if (!sentWords.contains(word)) {
                                    session.sendMessage(new TextMessage(ASR_PREFIX + word));
                                    logger.info("发送ASR消息:{}", ASR_PREFIX + word);
                                    sentWords.add(word);
                                }
                            }
                        }
                    }

                    if (!utterances.isEmpty()) {
                        JsonObject utterance = utterances.get(0).getAsJsonObject();
                        if (utterance.has("definite") && utterance.get("definite").isJsonPrimitive()) {
                            isFinal = utterance.get("definite").getAsBoolean();
                        }


                    }
                }

                // 处理有效结果
                if (!text.isEmpty()) {
                    //中断前端播放
                    session.sendMessage(new TextMessage(TTS_VOICE_STOP));
                    //logger.info("发送中断前端播放TTS_VOICE_STOP");
                    interruptTTS();
                    //logger.info("识别内容: {} | 是否最终结果: {}", text, isFinal);
                    if (isFinal) {
                        hasSentInputTextStart = false;
                        this.sentWords = new HashSet<>();
                        handleFinalResult(session, text);
                    }
                }
            } else {
                logger.warn("响应中缺少 result 字段");
            }

        } catch (Exception e) {
            logger.error("解析服务端响应失败", e);
        }
    }

    private static byte[] gzipDecompress(byte[] compressed) {
        try (ByteArrayInputStream bis = new ByteArrayInputStream(compressed);
             GZIPInputStream gzip = new GZIPInputStream(bis);
             ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
            byte[] buffer = new byte[1024];
            int len;
            while ((len = gzip.read(buffer)) > 0) {
                bos.write(buffer, 0, len);
            }
            return bos.toByteArray();
        } catch (IOException e) {
            logger.error("GZIP 解压失败", e);
            return new byte[0];
        }
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        logger.info("Connection closed by " + (remote ? "remote peer" : "us") + " Code: " + code + " Reason: " + reason);
    }

    @Override
    public void onError(Exception ex) {
        logger.error("VolcanoAsrClient error", ex);
    }

    // 新增重试控制变量
    private volatile boolean shouldRetry = true;
    private static final int MAX_RETRY_COUNT = 3;
    private int currentRetryCount = 0;

    public void interruptTTS() {
        if (ttsThread != null) {
            // 调用 TTS 线程的停止方法
            ttsThread.interruptTTS();
        }
    }


    // 增强的音频发送方法
    public void sendBytes(byte[] audioData) {
        // 缓存音频数据
        ConcurrentHashMap<String, Object> practiceCache = SessionManager.getPracticeCache(session);
        if (practiceCache != null) {
            practiceCache.put("cachedAudioData", audioData);
        }

        if (!isOpen()) {
            logger.warn("WebSocket 连接已断开，尝试重连");
            reconnectAndResend();
            return;
        }

        try {
            byte[] audioOnlyRequest = buildAudioOnlyRequest(audioData, false);
            this.send(audioOnlyRequest);
            currentRetryCount = 0; // 重置重试计数器
            //logger.info("成功发送音频数据: {}字节", audioData.length);
        } catch (Exception e) {
            logger.error("音频发送失败", e);
            handleSendError(audioData);
        }
    }


    // 修改错误处理方法
    private void handleSendError(byte[] audioData) {
        if (currentRetryCount < MAX_RETRY_COUNT) {
            logger.warn("准备重试发送({}/{}）", currentRetryCount + 1, MAX_RETRY_COUNT);
            try {
                TimeUnit.SECONDS.sleep(1);
                sendBytes(audioData);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        } else {
            logger.error("已达到最大重试次数，放弃发送");
            notifyFrontendError("音频传输失败，请检查网络连接");
        }
    }

    private void notifyFrontendError(String message) {
        try {
            JsonObject error = new JsonObject();
            error.addProperty("type", "error");
            error.addProperty("code", "ASR_TIMEOUT");
            error.addProperty("message", message);
            session.sendMessage(new TextMessage(error.toString()));
        } catch (IOException e) {
            logger.error("前端错误通知发送失败: {}", e.getMessage());
        }
    }

    private boolean handleConnectionState(byte[] audioData) {
        boolean isOpen = isOpen();
        if (!isOpen && shouldRetry && currentRetryCount < MAX_RETRY_COUNT) {
            logger.info("尝试重连({}/{})", currentRetryCount + 1, MAX_RETRY_COUNT);
            reconnectAndResend();
            currentRetryCount++;
            return false;
        }
        return true;
    }

    private VolcanoAsrClient createNewClientInstance() {
        return new VolcanoAsrClient(
                this.getURI(),
                this.session,
                this.redisService,
                this.webClientConfig,
                this.ttsUrl,
                this.saveVoicePath,
                this.ttsType,
                this.llmFlag,
                this.llmUrl
        );
    }

    private void reconnectAndResend() {
        try {
            // 关闭当前连接
            if (isOpen()) {
                this.close();
            }

            // 创建新的客户端实例
            VolcanoAsrClient newClient = createNewClientInstance();
            newClient.connectBlocking(); // 阻塞直到连接成功

            // 重新发送缓存的音频数据
            ConcurrentHashMap<String, Object> practiceCache = SessionManager.getPracticeCache(session);
            if (practiceCache != null) {
                byte[] cachedAudioData = (byte[]) practiceCache.get("cachedAudioData");
                if (cachedAudioData != null) {
                    newClient.sendBytes(cachedAudioData);
                    logger.info("成功重连并重新发送音频数据");
                } else {
                    logger.warn("未找到缓存的音频数据");
                }
            }

            // 更新会话缓存中的客户端实例
            if (practiceCache != null) {
                practiceCache.put(SessionManager.FUNASR_CACHE, newClient);
                logger.info("客户端实例已更新");
            }
        } catch (InterruptedException e) {
            logger.error("重连中断", e);
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            logger.error("重连失败", e);
            notifyFrontendError("重连失败，请检查网络连接");
        }
    }

    /**
     * 发送结束信号
     */
    public void sendEof() {
        if (!isOpen()) {
            logger.error("WebSocket connection is not open, cannot send EOF");
            return;
        }

        // 构建 AUDIO_ONLY_REQUEST 消息（结束信号）
        byte[] eofRequest = buildAudioOnlyRequest(new byte[0], true);

        // 发送结束信号
        this.send(eofRequest);
        logger.info("Sent EOF to Volcano ASR");
    }

    private void sendTypewriterEffect(WebSocketSession session, String text, long delayMillis) throws IOException, InterruptedException {
        for (char c : text.toCharArray()) {
            session.sendMessage(new TextMessage(String.valueOf(c))); // 逐字符发送
            Thread.sleep(delayMillis); // 添加延迟
        }
    }


    private void handleFinalResult(WebSocketSession session, String text) {
        try {
            logger.info("ASR识别结果:" + text);

            // 1. 发送ASR结果到页面-用户自己说的
            //session.sendMessage(new TextMessage(text));
            // 使用打字机效果发送文本 100ms 延迟
            //sendTypewriterEffect(session, text, 100);
            session.sendMessage(new TextMessage(inputTextStop));
            logger.info("发送ASR消息 inputTextStop");

            // 2. 更新对话上下文
            String sessionId = SessionManager.getSession(session).getNick();
            List<ChatCompletionMessage> history = loadChatHistoryInline(sessionId);

            ChatCompletionMessage userMessage = new ChatCompletionMessage();
            userMessage.setContent(text);
            userMessage.setRole(LLMRole.USER.getRole());
            history.add(userMessage);
            logger.info("更新对话上下文");
            // 3. 调用大模型服务
            Flux<String> responseStream;
            ObjectNode requestBody;
            switch (this.llmFlag) {
                case "LLM_CUSTOMER":
                    //logger.info("模拟客服开始");
                    //conversationId = ConversationUtil.startConversation();
                    logger.info("模拟客服开始，会话ID:{}", this.conversationId);
                    requestBody = buildLlmEngineRequest(session, text, this.conversationId);
                    responseStream = ConversationUtil.sendChatRequest(this.llmUrl, requestBody);
                    break;
                case "LLM_USER":
                    // 启动与客户的对话，获取 conversation_id
                    //logger.info("模拟客户开始");
                    //conversationId = ConversationUtil.startConversationWithCustomer();
                    logger.info("模拟客户开始，会话ID:{}", this.conversationId);
                    requestBody = buildLlmEngineRequest(session, text, this.conversationId);
                    responseStream = ConversationUtil.sendChatCustomerRequest(this.llmUrl, requestBody);
                    break;
                case "LLM_VOLCANO":
                default:
                    boolean stream = true; // 启用流式响应
                    responseStream = webClient.post()
                            .uri("https://ark.cn-beijing.volces.com/api/v3/chat/completions")
                            .header(HttpHeaders.AUTHORIZATION, "Bearer 631999ff-4583-4944-9511-da68d2a58911")
                            .contentType(MediaType.APPLICATION_JSON)
                            .bodyValue(buildVolcEngineRequest(session, text, stream))
                            .retrieve()
                            .bodyToFlux(String.class);
            }
            session.sendMessage(new TextMessage(replyTextStart));

            logger.info("发送TTS replyTextStart");
            // 4. 处理流式响应
            ttsThread = new ChatResponseThread(session, this.ttsUrl, this.saveVoicePath, this.ttsType, this.ttsClient, this.aliTtsClient);
            ttsThread.start();

            responseStream.subscribe(
                    data -> {
                        try {
                            if ("[DONE]".equals(data)) {
                                // 向前端发送文本停止消息
                                session.sendMessage(new TextMessage(replyTextStop));
                                logger.info("发送TTS replyTextStop");
                                handleStreamCompletion(session, ttsThread);
                                return;
                            }
                            ChatCompletionChunkResponse response = GSON.fromJson(data, ChatCompletionChunkResponse.class);
                            //logger.info("LLM原始响应:"+response.toString());
                            if (response != null &&
                                    response.getChoices() != null &&
                                    !response.getChoices().isEmpty() &&
                                    response.getChoices().get(0).getDelta() != null) {

                                //String content = response.getChoices().get(0).getDelta().getContent();
                                String content = response.getChoices().getFirst().getDelta().getContent().replace("\"", "");
                                if (content != null && !content.isEmpty()) {
                                    if (this.ttsType.equals("STEP") || this.ttsType.equals("ALI")) {
                                        ttsThread.sendTextToTTS(content);
                                        StringBuilder interfereCache = (StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE);
                                        interfereCache.append(content);
                                    } else {
                                        session.sendMessage(new TextMessage(TTS_PREFIX + content));
                                        StringBuilder interfereCache = (StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE);
                                        interfereCache.append(content);
                                        String tempTextCache = (String) SessionManager.getPracticeCache(session).get(TEMP_TEXT_CACHE);
                                        tempTextCache = tempTextCache + content;
                                        ConcurrentLinkedQueue<String> textQueue =
                                                (ConcurrentLinkedQueue<String>) SessionManager.getPracticeCache(session).get(SENTENCE_CACHE);
                                        //logger.info("[NOT DONE] tempTextCache：{}", tempTextCache);
                                        tempTextCache = sentenceSplit(SENTENCE_SPLIT_REGEX, tempTextCache, textQueue);
                                        //logger.info("[NOT DONE] sentenceCache：{}, tempTextCache:{}", textQueue, tempTextCache);
                                        SessionManager.getPracticeCache(session).put(TEMP_TEXT_CACHE, tempTextCache);
                                        //logger.info("subscribe content:{}, interfereCache:{}, textQueue:{}", content, interfereCache, textQueue);
                                    }

                                }
                            }

                        } catch (Exception e) {
                            logger.error("流数据处理异常: {}", e.getMessage());
                        }
                    },
                    error -> {
                        logger.error("流请求失败: {}", error.getMessage());
                        ttsThread.setStop();
                    },
                    () -> {
                        try {
                            session.sendMessage(new TextMessage(replyTextStop));
                            session.sendMessage(new TextMessage(replyAudioStop));
                        } catch (IOException e) {
                            logger.error("结束消息发送失败: {}", e.getMessage());
                        }
                    }
            );

        } catch (Exception e) {
            logger.error("最终结果处理失败: {}", e.getMessage());
            try {
                session.sendMessage(new TextMessage("ASR处理错误:" + e.getMessage()));
            } catch (IOException ioEx) {
                logger.error("错误消息发送失败: {}", ioEx.getMessage());
            }
        }
    }


    public void handleCustomer(WebSocketSession session) {
        try {
            // 3. 调用大模型服务
            Flux<String> responseStream = ConversationUtil.sendPrologueRequest(this.llmUrl, this.conversationId);
            session.sendMessage(new TextMessage(replyTextStart));

            logger.info("发送TTS replyTextStart");
            // 4. 处理流式响应
            ttsThread = new ChatResponseThread(session, this.ttsUrl, this.saveVoicePath, this.ttsType, this.ttsClient, this.aliTtsClient);
            ttsThread.start();

            responseStream.subscribe(
                    data -> {
                        try {
                            if ("[DONE]".equals(data)) {
                                // 向前端发送文本停止消息
                                session.sendMessage(new TextMessage(replyTextStop));
                                logger.info("发送TTS replyTextStop");
                                handleStreamCompletion(session, ttsThread);
                                return;
                            }
                            ChatCompletionChunkResponse response = GSON.fromJson(data, ChatCompletionChunkResponse.class);
                            if (response != null &&
                                    response.getChoices() != null &&
                                    !response.getChoices().isEmpty() &&
                                    response.getChoices().get(0).getDelta() != null) {

                                String content = response.getChoices().getFirst().getDelta().getContent().replace("\"", "");
                                if (content != null && !content.isEmpty()) {
                                    if (this.ttsType.equals("STEP")) {
                                        ttsThread.sendTextToTTS(content);
                                        StringBuilder interfereCache = (StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE);
                                        interfereCache.append(content);
                                    } else {
                                        session.sendMessage(new TextMessage(FIRST_TTS_PREFIX + content));
                                        StringBuilder interfereCache = (StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE);
                                        interfereCache.append(content);
                                        String tempTextCache = (String) SessionManager.getPracticeCache(session).get(TEMP_TEXT_CACHE);
                                        tempTextCache = tempTextCache + content;
                                        ConcurrentLinkedQueue<String> textQueue =
                                                (ConcurrentLinkedQueue<String>) SessionManager.getPracticeCache(session).get(SENTENCE_CACHE);
                                        //logger.info("[NOT DONE] tempTextCache：{}", tempTextCache);
                                        tempTextCache = sentenceSplit(SENTENCE_SPLIT_REGEX, tempTextCache, textQueue);
                                        //logger.info("[NOT DONE] sentenceCache：{}, tempTextCache:{}", textQueue, tempTextCache);
                                        SessionManager.getPracticeCache(session).put(TEMP_TEXT_CACHE, tempTextCache);
                                        //logger.info("subscribe content:{}, interfereCache:{}, textQueue:{}", content, interfereCache, textQueue);
                                    }

                                }
                            }

                        } catch (Exception e) {
                            logger.error("流数据处理异常: {}", e.getMessage());
                        }
                    },
                    error -> {
                        logger.error("流请求失败: {}", error.getMessage());
                        ttsThread.setStop();
                    },
                    () -> {
                        try {
                            session.sendMessage(new TextMessage(replyTextStop));
                            session.sendMessage(new TextMessage(replyAudioStop));
                        } catch (IOException e) {
                            logger.error("结束消息发送失败: {}", e.getMessage());
                        }
                    }
            );

        } catch (Exception e) {
            logger.error("最终结果处理失败: {}", e.getMessage());
            try {
                session.sendMessage(new TextMessage("ASR处理错误:" + e.getMessage()));
            } catch (IOException ioEx) {
                logger.error("错误消息发送失败: {}", ioEx.getMessage());
            }
        }
    }


    private List<ChatCompletionMessage> loadChatHistoryInline(String sessionId) {
        String historyJson = redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId));
        return historyJson != null ?
                GSON.fromJson(historyJson, new TypeToken<List<ChatCompletionMessage>>() {
                }.getType()) :
                new ArrayList<>();
    }

    private ObjectNode buildLlmEngineRequest(WebSocketSession session, String text, String conversationId) {
        ObjectMapper objectMapper = new ObjectMapper();
        ObjectNode request = objectMapper.createObjectNode();
        request.put("conversation_id", conversationId);

        ArrayNode messages = objectMapper.createArrayNode();

        // 内联历史记录加载
        String sessionId = SessionManager.getSession(session).getNick();
        List<ChatCompletionMessage> history = GSON.fromJson(
                redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId)),
                new TypeToken<List<ChatCompletionMessage>>() {
                }.getType()
        );
        if (history == null) {
            history = new ArrayList<>();
        }

        // 添加历史消息
        for (ChatCompletionMessage msg : history) {
            ObjectNode message = objectMapper.createObjectNode();
            String role = msg.getRole();
            if (role == null || role.trim().isEmpty()) {
                // 默认角色为 "user"
                role = LLMRole.USER.getRole();
            }
            message.put("role", role.toLowerCase());
            message.put("content", msg.getContent());
            messages.add(message);
        }

        // 添加当前消息
        ObjectNode currentMessage = objectMapper.createObjectNode();
        currentMessage.put("role", LLMRole.USER.getRole().toLowerCase());
        currentMessage.put("content", text);
        messages.add(currentMessage);

        // 校验messages是否为空
        if (messages.size() == 0) {
            throw new IllegalArgumentException("Messages cannot be empty");
        }

        request.set("messages", messages);
        return request;
    }

    private ObjectNode buildVolcEngineRequest(WebSocketSession session, String text, boolean stream) {
        ObjectMapper objectMapper = new ObjectMapper();
        ObjectNode request = objectMapper.createObjectNode();
        request.put("model", "ep-20250117095237-vzjjf"); // 可以配置为动态参数
        request.put("stream", stream); // 动态设置流式响应

        ArrayNode messages = objectMapper.createArrayNode();

        // 内联历史记录加载
        String sessionId = SessionManager.getSession(session).getNick();
        List<ChatCompletionMessage> history = GSON.fromJson(
                redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId)),
                new TypeToken<List<ChatCompletionMessage>>() {
                }.getType()
        );
        if (history == null) {
            history = new ArrayList<>();
        }

        // 添加历史消息
        for (ChatCompletionMessage msg : history) {
            ObjectNode message = objectMapper.createObjectNode();
            String role = msg.getRole();
            if (role == null || role.trim().isEmpty()) {
                // 默认角色为 "user"
                role = LLMRole.USER.getRole();
            }
            message.put("role", role.toLowerCase());
            message.put("content", msg.getContent());
            messages.add(message);
        }

        // 添加当前消息
        ObjectNode currentMessage = objectMapper.createObjectNode();
        currentMessage.put("role", LLMRole.USER.getRole().toLowerCase());
        currentMessage.put("content", text);
        messages.add(currentMessage);

        // 校验messages是否为空
        if (messages.size() == 0) {
            throw new IllegalArgumentException("Messages cannot be empty");
        }

        request.set("messages", messages);
        return request;
    }


    private JsonObject createMessageObject(String role, String content) {
        JsonObject message = new JsonObject();
        message.addProperty("role", role);
        message.addProperty("content", content);
        return message;
    }

    // 复用原有方法
    private String sentenceSplit(String regex, String text, ConcurrentLinkedQueue<String> queue) {
        Pattern pattern = Pattern.compile(regex);
        Matcher matcher = pattern.matcher(text);
        int lastPos = 0;
        StringBuilder buffer = new StringBuilder();

        while (matcher.find()) {
            buffer.append(text, lastPos, matcher.end());
            lastPos = matcher.end();

            if (buffer.length() > SENTENCE_SEND_MIN_WORDS_LIMIT) {
                queue.add(buffer.toString());
                buffer.setLength(0);
            }
        }
        return buffer.append(text.substring(lastPos)).toString();
    }

    // 复用原有逻辑处理完成状态
    private void handleStreamCompletion(WebSocketSession session, ChatResponseThread ttsThread) {
        try {
            // 保存助理响应
            String sessionId = SessionManager.getSession(session).getNick();
            List<ChatCompletionMessage> history = GSON.fromJson(
                    redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId)),
                    new TypeToken<List<ChatCompletionMessage>>() {
                    }.getType()
            );
            history.add(new ChatCompletionMessage(
                    LLMRole.ASSISTANT,
                    ((StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE)).toString()
            ));
            redisService.set(
                    String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId),
                    GSON.toJson(history),
                    7, TimeUnit.DAYS
            );

            // 处理残留文本
            String tempText = (String) SessionManager.getPracticeCache(session).get(TEMP_TEXT_CACHE);
            if (StringUtils.isNotEmpty(tempText)) {
                ((ConcurrentLinkedQueue<String>) SessionManager.getPracticeCache(session).get(SENTENCE_CACHE)).add(tempText);
            }
            SessionManager.getPracticeCache(session).put(TEMP_TEXT_CACHE, "");

            // 停止TTS线程
            ttsThread.setStop();
        } catch (Exception e) {
            logger.error("流完成处理异常: {}", e.getMessage());
        }
    }


    /**
     * 发送 FULL_CLIENT_REQUEST 消息
     */
    private void sendFullClientRequest() {
        // 构建 FULL_CLIENT_REQUEST 消息
        JsonObject user = new JsonObject();
        user.addProperty("uid", "test");

        JsonObject audio = new JsonObject();
        audio.addProperty("format", "pcm");
        audio.addProperty("sample_rate", 16000);
        audio.addProperty("bits", 16);
        audio.addProperty("channel", 1);
        audio.addProperty("codec", "raw");

        JsonObject request = new JsonObject();
        request.addProperty("model_name", "bigmodel");
        //单位ms，默认为3000。当静音时间超过该值时，会将文本分为两个句子。不决定判停，所以不会修改definite出现的位置。在end_window_size配置后，该参数失效。
        //request.addProperty("vad_segment_duration", 500);
        //单位ms，默认为800，最小200。静音时长超过该值，会直接判停，输出definite。配置该值，不使用语义分句，根据静音时长来分句。用于实时性要求较高场景，可以提前获得definite句子
        //request.addProperty("end_window_size", 800);
        //默认为false。**语义顺滑**‌是一种技术，旨在提高自动语音识别（ASR）结果的文本可读性和流畅性。这项技术通过删除或修改ASR结果中的不流畅部分，如停顿词、语气词、语义重复词等，使得文本更加易于阅读和理解。
        request.addProperty("enable_ddc", true);
        // 启用标点
        request.addProperty("enable_punc", true);
        // 设置为"single"则为增量结果返回，即不返回之前分句的结果。
        request.addProperty("result_type", "single");
        request.addProperty("show_utterances", true);

        JsonObject payload = new JsonObject();
        payload.add("user", user);
        payload.add("audio", audio);
        payload.add("request", request);

        String payloadStr = payload.toString();
        logger.info("FULL_CLIENT_REQUEST payload: {}", payloadStr);

        // 压缩 payload
        byte[] payloadBytes = gzipCompress(payloadStr.getBytes(StandardCharsets.UTF_8));

        // 构建消息头
        byte[] header = getHeader(FULL_CLIENT_REQUEST, POS_SEQUENCE, JSON, GZIP, (byte) 0);

        // 构建序列号
        int seq = sequence++;
        byte[] seqBytes = intToBytes(seq);

        // 构建 payload 大小
        byte[] payloadSize = intToBytes(payloadBytes.length);

        // 组装 FULL_CLIENT_REQUEST 消息
        byte[] fullClientRequest = new byte[header.length + seqBytes.length + payloadSize.length + payloadBytes.length];
        int destPos = 0;
        System.arraycopy(header, 0, fullClientRequest, destPos, header.length);
        destPos += header.length;
        System.arraycopy(seqBytes, 0, fullClientRequest, destPos, seqBytes.length);
        destPos += seqBytes.length;
        System.arraycopy(payloadSize, 0, fullClientRequest, destPos, payloadSize.length);
        destPos += payloadSize.length;
        System.arraycopy(payloadBytes, 0, fullClientRequest, destPos, payloadBytes.length);

        // 发送 FULL_CLIENT_REQUEST 消息
        this.send(fullClientRequest);
        logger.info("Sent FULL_CLIENT_REQUEST to Volcano ASR");
    }

    /**
     * 构建 AUDIO_ONLY_REQUEST 消息
     *
     * @param audioData 音频数据
     * @param isEof     是否为结束信号
     * @return AUDIO_ONLY_REQUEST 消息
     */
    private byte[] buildAudioOnlyRequest(byte[] audioData, boolean isEof) {
        // 构建消息头
        byte messageTypeSpecificFlags = isEof ? NEG_WITH_SEQUENCE : POS_SEQUENCE;
        byte[] header = getHeader(AUDIO_ONLY_REQUEST, messageTypeSpecificFlags, JSON, GZIP, (byte) 0);

        // 构建序列号
        int seq = sequence++;
        if (isEof) {
            seq = -seq; // 结束信号的序列号为负数
        }
        byte[] seqBytes = intToBytes(seq);

        // 压缩音频数据
        byte[] payloadBytes = gzipCompress(audioData);

        // 构建 payload 大小
        byte[] payloadSize = intToBytes(payloadBytes.length);

        // 组装 AUDIO_ONLY_REQUEST 消息
        byte[] audioOnlyRequest = new byte[header.length + seqBytes.length + payloadSize.length + payloadBytes.length];
        int destPos = 0;
        System.arraycopy(header, 0, audioOnlyRequest, destPos, header.length);
        destPos += header.length;
        System.arraycopy(seqBytes, 0, audioOnlyRequest, destPos, seqBytes.length);
        destPos += seqBytes.length;
        System.arraycopy(payloadSize, 0, audioOnlyRequest, destPos, payloadSize.length);
        destPos += payloadSize.length;
        System.arraycopy(payloadBytes, 0, audioOnlyRequest, destPos, payloadBytes.length);

        return audioOnlyRequest;
    }

    /**
     * 构建消息头
     *
     * @param messageType              消息类型
     * @param messageTypeSpecificFlags 消息类型特定标志
     * @param serialMethod             序列化方法
     * @param compressionType          压缩类型
     * @param reservedData             保留数据
     * @return 消息头
     */
    private static byte[] getHeader(byte messageType, byte messageTypeSpecificFlags, byte serialMethod, byte compressionType, byte reservedData) {
        final byte[] header = new byte[4];
        header[0] = (byte) ((PROTOCOL_VERSION << 4) | DEFAULT_HEADER_SIZE); // Protocol version | header size
        header[1] = (byte) ((messageType << 4) | messageTypeSpecificFlags); // Message type | messageTypeSpecificFlags
        header[2] = (byte) ((serialMethod << 4) | compressionType); // Serialization method | compression type
        header[3] = reservedData; // Reserved data
        return header;
    }

    /**
     * 将整数转换为字节数组
     *
     * @param value 整数
     * @return 字节数组
     */
    private static byte[] intToBytes(int value) {
        return new byte[]{
                (byte) ((value >> 24) & 0xFF),
                (byte) ((value >> 16) & 0xFF),
                (byte) ((value >> 8) & 0xFF),
                (byte) (value & 0xFF)
        };
    }

    /**
     * 压缩数据
     *
     * @param data 原始数据
     * @return 压缩后的数据
     */
    private static byte[] gzipCompress(byte[] data) {
        try (ByteArrayOutputStream out = new ByteArrayOutputStream();
             GZIPOutputStream gzip = new GZIPOutputStream(out)) {
            gzip.write(data);
            gzip.finish();
            return out.toByteArray();
        } catch (IOException e) {
            logger.error("Failed to compress data", e);
            return new byte[0];
        }
    }


}
